package com.jy.oms.framework.asyn.task.queue.database;

import com.alibaba.fastjson.JSON;
import com.jy.oms.framework.asyn.task.checker.model.ExecutorCheckResult;
import com.jy.oms.framework.asyn.task.cluster.config.ITaskConfigChangeListener;
import com.jy.oms.framework.asyn.task.cluster.config.ITaskConfigService;
import com.jy.oms.framework.asyn.task.constants.CacheConstants;
import com.jy.oms.framework.asyn.task.constants.ConfigConstants;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.constants.TaskStateEnum;
import com.jy.oms.framework.asyn.task.core.ExecutorConfig;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.core.RedissonClientProxy;
import com.jy.oms.framework.asyn.task.core.TaskIdCreateService;
import com.jy.oms.framework.asyn.task.exception.TaskConfigException;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerProcessor;
import com.jy.oms.framework.asyn.task.model.TaskConfig;
import com.jy.oms.framework.asyn.task.model.TaskDto;
import com.jy.oms.framework.asyn.task.model.TaskGroup;
import com.jy.oms.framework.asyn.task.model.TaskHandlerWrapper;
import com.jy.oms.framework.asyn.task.model.TaskQueryDto;
import com.jy.oms.framework.asyn.task.model.TaskResultDto;
import com.jy.oms.framework.asyn.task.model.TaskWrapper;
import com.jy.oms.framework.asyn.task.model.strategy.ExecuteStrategyObj;
import com.jy.oms.framework.asyn.task.queue.AbstractTaskQueueManager;
import com.jy.oms.framework.asyn.task.queue.database.entity.SysTaskQuery;
import com.jy.oms.framework.asyn.task.queue.database.entity.SysTaskQueue;
import com.jy.oms.framework.asyn.task.queue.database.service.ISysTaskQueueService;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

@Service
public class DbTaskQueueManager extends AbstractTaskQueueManager implements ITaskHandlerProcessor {
    @Autowired
    private ISysTaskQueueService sysTaskQueueService;
    @Autowired
    private TaskIdCreateService idCreateService;
    @Autowired
    private ITaskHandlerManager taskHandlerManager;
    @Autowired
    private ITaskConfigService configService;
    private int startTimeHour = 2;

    @Override
    public void processAfterTaskHandlerInitialization() {
        //监听配置变化
        configService.addTaskConfigChangeListener(new ITaskConfigChangeListener() {
            @Override
            public String getConfigKey() {
                return ConfigConstants.EXTRACTOR_DB_QUEUE_QUERY_START_HOUR;
            }

            @Override
            public void onConfigChange(String configKey, String config) {
                if (NumberUtils.isCreatable(config)) {
                    int hour = Integer.parseInt(config);
                    if (hour > 0) {
                        startTimeHour = hour;
                    }
                }
            }
        });
    }

    private String getTaskQueueName(String taskCode) {
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskCode);
        if (!taskHandlerWrapper.isEnableExecute()) {
            if (taskHandlerWrapper.isEnableAddTaskIfUnableExecute()) {
                return this.getTaskQueueNameByGroup(taskHandlerWrapper.getDefaultTaskGroupIfUnableExecute());
            }
            return null;
        }
        ExecuteStrategyObj executeStrategy = taskHandlerWrapper.getExecuteStrategy();
        return StringUtils.defaultString(executeStrategy.getTaskGroupDbQueueName(), "");
    }

    private String getTaskQueueNameByGroup(String taskGroup) {
        Map<TaskGroup, List<TaskHandlerWrapper>> taskGroupListMap = taskHandlerManager.loadAllExecutableTaskHandler();
        List<TaskHandlerWrapper> taskHandlerWrapperList = taskGroupListMap.entrySet().stream()
                .filter(entry -> entry.getKey().getTaskQueueType() == TaskQueueTypeEnum.DB)
                .filter(entry -> entry.getKey().getGroup().equals(taskGroup))
                .map(Map.Entry::getValue)
                .findFirst().orElse(null);
        if (CollectionUtils.isEmpty(taskHandlerWrapperList)) {
            return null;
        }

        ExecuteStrategyObj executeStrategy = taskHandlerWrapperList.get(0).getExecuteStrategy();
        return StringUtils.defaultString(executeStrategy.getTaskGroupDbQueueName(), "");
    }

    private boolean judgeKeepUniqueTaskAccordParam(String taskCode) {
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskCode);
        TaskConfig taskConfig = taskHandlerWrapper.getTaskConfig();
        return taskConfig.isKeepUniqueTaskAccordParam();
    }

    @Override
    public void add(TaskWrapper taskWrapper) {
        super.add(taskWrapper);

        String taskCode = taskWrapper.getTaskCode();
        String taskQueueName = this.getTaskQueueName(taskCode);
        if (taskQueueName == null) {
            throw new TaskConfigException("不能添加任务，taskCode：" + taskCode);
        }

        SysTaskQueue sysTaskQueue = this.convertSysTask(taskWrapper);
        if (this.judgeKeepUniqueTaskAccordParam(taskCode)) {
            String cacheKey = CacheConstants.genCacheKey(CacheConstants.UNIQUE_TASK_ACCOARD_PARAM_LOCK, sysTaskQueue.getUuid());
            RedissonClientProxy.lock(cacheKey, () -> {
                SysTaskQueue oldSysTaskQueue = this.querySysTaskQueue(taskWrapper, sysTaskQueue.getUuid(), null);
                if (oldSysTaskQueue == null) {
                    sysTaskQueueService.insertSysTask(sysTaskQueue, taskQueueName);
                }

                return null;
            });
        } else {
            sysTaskQueueService.insertSysTask(sysTaskQueue, taskQueueName);
        }
    }

    @Transactional
    @Override
    public void addBatch(List<TaskWrapper> taskWrapperList) {
        super.addBatch(taskWrapperList);

        Map<String, List<TaskWrapper>> taskWrapperMap = taskWrapperList.stream()
                .collect(Collectors.groupingBy(e -> this.getTaskQueueName(e.getTaskCode())));
        for (Map.Entry<String, List<TaskWrapper>> entry : taskWrapperMap.entrySet()) {
            List<SysTaskQueue> sysTaskList = entry.getValue().stream().map(this::convertSysTask).collect(Collectors.toList());
            sysTaskQueueService.insertSysTaskList(sysTaskList, entry.getKey());
        }
    }

    private SysTaskQueue convertSysTask(TaskWrapper taskWrapper) {
        ExecutorConfig executorConfig = ExecutorContext.getExecutorConfig();
        TaskDto taskDto = taskWrapper.getTaskDto();

        SysTaskQueue sysTask = new SysTaskQueue();
        sysTask.setId(idCreateService.getId());
        sysTask.setBusinessGroup(executorConfig.getBusinessGroup());
        sysTask.setEnvFlag(executorConfig.getEnv());
        sysTask.setTaskGroup(taskWrapper.getTaskGroup());
        sysTask.setTaskCode(taskDto.getTaskCode());
        sysTask.setBillCode(taskDto.getBillCode());

        String taskParam = this.convertTaskParam(taskDto);
        sysTask.setTaskParam(taskParam);
        sysTask.setUuid(this.generateTaskIdentifier(taskDto.getTaskCode(), taskParam));

        sysTask.setCallCount(0);
        sysTask.setState(taskDto.getTaskState() == null ? TaskStateEnum.UNDO.getCode() : taskDto.getTaskState().getCode());

        if (taskDto.getDelayedSeconds() != null) {
            sysTask.setEarliestExecuteTime(DateUtils.addSeconds(new Date(), taskDto.getDelayedSeconds().intValue()));
        } else {
            sysTask.setEarliestExecuteTime(new Date());
        }
        sysTask.setCreateDate(new Date());
        sysTask.setCreateBy(StringUtils.defaultIfBlank(taskDto.getCreator(), this.getServerIp()));
        return sysTask;
    }

    @Override
    public List<TaskWrapper> poll(TaskQueryDto taskQuery) {
        ExecutorConfig executorConfig = ExecutorContext.getExecutorConfig();
        SysTaskQuery query = new SysTaskQuery();
        query.setBusinessGroup(executorConfig.getBusinessGroup());
        query.setEnvFlag(executorConfig.getEnv());
        query.setTaskGroup(taskQuery.getTaskGroup());
        query.setTaskCodes(taskQuery.getTaskCodes());
        query.setExcludeTaskCodes(taskQuery.getExcludeTaskCodes());
        query.setState(TaskStateEnum.UNDO.getCode());
        query.setPageSize(taskQuery.getTaskSizePerQuery());

        Date endTime = taskQuery.getEndTime() != null ? taskQuery.getEndTime() : new Date();
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(endTime);
        calendar.add(Calendar.HOUR_OF_DAY, -startTimeHour);
        Date startTime = calendar.getTime();
        query.setStartTime(startTime);
        query.setEndTime(endTime);

        //根据任务最早执行时间升序排序，每次最多查询200条
        String taskQueueName = this.getTaskQueueNameByGroup(taskQuery.getTaskGroup());
        List<SysTaskQueue> result = sysTaskQueueService.querySysTaskList(query, taskQueueName);
        if (CollectionUtils.isEmpty(result)) {
            return Collections.emptyList();
        }

        return result.stream().map(sysTask -> {
            Object param = this.getTaskParam(sysTask.getTaskCode(), sysTask.getTaskParam(), sysTask.getUuid());
            if (param == null) {
                return null;
            }

            TaskDto taskDto = new TaskDto();
            taskDto.setTaskCode(sysTask.getTaskCode());
            taskDto.setParam(param);
            taskDto.setBillCode(sysTask.getBillCode());

            TaskWrapper taskWrapper = new TaskWrapper();
            taskWrapper.setTaskCode(sysTask.getTaskCode());
            taskWrapper.setTaskDto(taskDto);
            taskWrapper.setTaskGroup(sysTask.getTaskGroup());
            taskWrapper.setTaskQueueId(sysTask.getId().toString());
            taskWrapper.setCallCount(sysTask.getCallCount());
            taskWrapper.setParamIdentifier(sysTask.getUuid());
            taskWrapper.setEarliestExecuteTime(sysTask.getEarliestExecuteTime());
            return taskWrapper;
        }).filter(Objects::nonNull).collect(Collectors.toList());
    }

    @Override
    public void finish(TaskResultDto taskResultDto) {
        super.finish(taskResultDto);

        if (taskResultDto.getTaskWrapper().getTaskQueueId() != null) {
            TaskWrapper taskWrapper = taskResultDto.getTaskWrapper();
            SysTaskQueue updateSysTask = this.convert(taskResultDto);
            sysTaskQueueService.updateById(updateSysTask, this.getTaskQueueName(taskWrapper.getTaskCode()));
        }
    }

    private SysTaskQueue convert(TaskResultDto taskResultDto) {
        TaskWrapper taskWrapper = taskResultDto.getTaskWrapper();
        SysTaskQueue updateSysTask = new SysTaskQueue();
        updateSysTask.setId(Long.valueOf(taskWrapper.getTaskQueueId()));

        if (TaskStateEnum.DONE == taskResultDto.getTaskState()) {
            updateSysTask.setState(TaskStateEnum.DONE.getCode());
        } else {
            if (taskResultDto.isCheckFilter()) {
                ExecutorCheckResult executorCheckResult = taskResultDto.getExecutorCheckResult();
                if (executorCheckResult.getEarliestExecuteTime() != null) {
                    updateSysTask.setEarliestExecuteTime(executorCheckResult.getEarliestExecuteTime());
                } else if (TaskStateEnum.DONE == executorCheckResult.getTaskState()) {
                    updateSysTask.setState(TaskStateEnum.DONE.getCode());
                }
            } else if (taskResultDto.getTaskState() == TaskStateEnum.INVALID) {
                //无效任务延时1天
                updateSysTask.setEarliestExecuteTime(DateUtils.addDays(new Date(), 1));
            } else if (taskResultDto.getTaskState() == TaskStateEnum.TIMEOUT) {
                //超时任务延时1小时(超时并不是没执行成功，延时时间不能小于[TaskRunningStateCollector]运行状态过期时间)
                //最多重试1次
                updateSysTask.setEarliestExecuteTime(DateUtils.addMinutes(new Date(), 61));
                if (taskResultDto.getCallCount() > 1) {
                    updateSysTask.setState(TaskStateEnum.TIMEOUT.getCode());
                }

                this.alarmTimeoutTask(taskResultDto);
            } else if (taskResultDto.isEnableRetry()) {
                //重试任务
                int retryIntervalSeconds = taskResultDto.isRetryBackoff() 
                        ? (Math.min(taskResultDto.getCallCount(), Math.max(taskResultDto.getRetryNumConfig(), 1))) * taskResultDto.getRetryIntervalSeconds() 
                        : taskResultDto.getRetryIntervalSeconds();
                updateSysTask.setEarliestExecuteTime(DateUtils.addSeconds(new Date(), retryIntervalSeconds));
            } else {
                updateSysTask.setState(taskResultDto.getTaskState().getCode());
            }

            if (taskResultDto.isUpdateTaskParamWhenException()) {
                if (taskWrapper.getTaskDto().getParam() instanceof String) {
                    updateSysTask.setTaskParam((String) taskWrapper.getTaskDto().getParam());
                } else {
                    updateSysTask.setTaskParam(JSON.toJSONString(taskWrapper.getTaskDto().getParam()));
                }
            }
        }

        updateSysTask.setCallCount(taskResultDto.getCallCount());
        updateSysTask.setUpdateBy(this.getServerIp());
        updateSysTask.setUpdateDate(new Date());
        return updateSysTask;
    }

    @Override
    public void finish(List<TaskResultDto> taskResultList) {
        super.finish(taskResultList);

        Map<String, List<TaskResultDto>> taskResultMap = taskResultList.stream()
                .collect(Collectors.groupingBy(e -> this.getTaskQueueName(e.getTaskWrapper().getTaskCode())));
        for (Map.Entry<String, List<TaskResultDto>> entry : taskResultMap.entrySet()) {
            List<SysTaskQueue> sysTaskList = entry.getValue().stream().map(this::convert).collect(Collectors.toList());
            sysTaskQueueService.updateBatchById(sysTaskList, entry.getKey());
        }
    }

    @Override
    public void remove(TaskWrapper taskWrapper) {
        TaskDto taskDto = taskWrapper.getTaskDto();
        String taskParam = this.convertTaskParam(taskDto);
        String uuid = this.generateTaskIdentifier(taskDto.getTaskCode(), taskParam);
        logger.info("删除DB任务, taskGroup={}, taskCode={}, paramIdentifier={}", taskWrapper.getTaskGroup(), taskDto.getTaskCode(), uuid);

        SysTaskQueue sysTaskQueue = this.querySysTaskQueue(taskWrapper, uuid, TaskStateEnum.UNDO);
        if (sysTaskQueue == null) {
            return;
        }

        SysTaskQueue updateSysTask = new SysTaskQueue();
        updateSysTask.setId(sysTaskQueue.getId());
        updateSysTask.setState(TaskStateEnum.DONE.getCode());
        updateSysTask.setUpdateBy(this.getServerIp() + "_remove");
        updateSysTask.setUpdateDate(new Date());

        String taskQueueName = this.getTaskQueueName(taskWrapper.getTaskCode());
        sysTaskQueueService.updateById(updateSysTask, taskQueueName);
    }

    private SysTaskQueue querySysTaskQueue(TaskWrapper taskWrapper, String uuid, TaskStateEnum taskState) {
        ExecutorConfig executorConfig = ExecutorContext.getExecutorConfig();
        TaskDto taskDto = taskWrapper.getTaskDto();
        SysTaskQuery sysTaskQuery = new SysTaskQuery();
        sysTaskQuery.setBusinessGroup(executorConfig.getBusinessGroup());
        sysTaskQuery.setEnvFlag(executorConfig.getEnv());
        sysTaskQuery.setTaskGroup(taskWrapper.getTaskGroup());
        sysTaskQuery.setTaskCode(taskDto.getTaskCode());
        sysTaskQuery.setPageSize(1);
        sysTaskQuery.setUuid(uuid);
        if (taskState != null) {
            sysTaskQuery.setState(taskState.getCode());
        }

        String taskQueueName = this.getTaskQueueName(taskWrapper.getTaskCode());
        if (taskQueueName == null) {
            return null;
        }

        List<SysTaskQueue> result = sysTaskQueueService.querySysTaskList(sysTaskQuery, taskQueueName);
        if (CollectionUtils.isEmpty(result)) {
            return null;
        }

        return result.get(0);
    }

}
